package com.s.rpc.core;

import com.s.rpc.channel.ClientChannel;
import com.s.rpc.serializer.Serializer;
import com.s.rpc.domain.RpcRequest;
import com.s.rpc.domain.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Data;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author ShenshenLuo
 * @email luoshenshen@buaa.edu.cn
 * @date 2025/5/11 13:40
 **/
@Data
@SuppressWarnings("all")
public class NettyClient {

    private String host;

    private int port;

    private Serializer serializer;

    private Channel channel;

    private final AtomicLong requestIdGenerator = new AtomicLong(0);

    private final ConcurrentHashMap<String, CompletableFuture<RpcResponse>> pendingRequests = new ConcurrentHashMap<>();

    public NettyClient() {
    }

    public NettyClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public NettyClient(String host, int port, Serializer serializer) {
        this.host = host;
        this.port = port;
        this.serializer = serializer;
    }

    public void startClient() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ClientChannel(pendingRequests, serializer));

            ChannelFuture future = bootstrap.connect(host, port).sync();
            channel = future.channel();
        } finally {
            // 这里不关闭group，因为要保持连接
        }
    }

    public CompletableFuture<RpcResponse> sendRequest(RpcRequest request) {
        CompletableFuture<RpcResponse> future = new CompletableFuture<>();
        pendingRequests.put(request.getRequestId(), future);
        channel.writeAndFlush(request).addListener((ChannelFutureListener) f -> {
            if (!f.isSuccess()) {
                future.completeExceptionally(f.cause());
                pendingRequests.remove(request.getRequestId());
            }
        });
        return future;
    }

    public <T> CompletableFuture<T> invoke(String className, String methodName,
                                           Class<?>[] parameterTypes, Object[] parameters) {
        RpcRequest request = new RpcRequest();
        request.setRequestId(String.valueOf(requestIdGenerator.incrementAndGet()));
        request.setClassName(className);
        request.setMethodName(methodName);
        request.setParameterTypes(parameterTypes);
        request.setParameters(parameters);

        return sendRequest(request).thenApply(response -> {
            return (T) response.getData();
        });
    }

    public void close() {
        if (channel != null) {
            channel.close();
        }
    }
}
